package com.fc.push.handler.pub.exec.task;

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttPersistenceException;

import com.fc.push.exception.ExcetionLevelConst;
import com.fc.push.exception.MServerException;
import com.fc.push.handler.pub.service.MqttPublishUtil;
import com.fc.push.mqtt.SharedMqttClientPoolFacade;

public class MqttPublishTask implements Runnable {
	private String appkey = null;
	private MqttMessage payload = null;
	private Object[] objects = null;

	public MqttPublishTask(String appkey,MqttMessage payload, Object... objs) {
		this.appkey = appkey;
		this.payload = payload;
		this.objects = objs;
	}

	@Override
	public void run() {
		MqttClient client = SharedMqttClientPoolFacade.getMqttClient();
		if(!client.isConnected()){
			try {
				client = SharedMqttClientPoolFacade.getMqttClient();
			} catch (Exception e) {
				throw new MServerException(e, e.getMessage(),ExcetionLevelConst.EXCEPTION_LEVEL_2);
			}
		}
		
		for (Object object : objects) {
			try {
				client.publish(MqttPublishUtil.assClientId(this.appkey,object.toString()), payload);
			} catch (MqttPersistenceException e) {
				throw new MServerException(e, e.getMessage(),ExcetionLevelConst.EXCEPTION_LEVEL_2);
			} catch (MqttException e) {
				throw new MServerException(e, e.getMessage(),ExcetionLevelConst.EXCEPTION_LEVEL_2);
			}
		}
		
		//关闭链接
		try {
			SharedMqttClientPoolFacade.returnResource(client);
		} catch (Exception e) {
			throw new MServerException(e, e.getMessage());
		}finally{
			client = null;
		}
	}
}
